package com.hu.flink12.api.wm;

import com.hu.flink12.api.entity.Order;
import com.hu.flink12.api.source.CustomerOrder;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

/**
 * @Author: hujianjun
 * @Date: 2021/2/6 21:59
 * @Describe:
 */
public class OrderWatermarkerTest {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /**
         * 重启策略：
         * 1.默认无限制重启
         * 2.无重启策略
         * 3.固定延迟重启策略  --开发中常用此策略
         * 4.失败率重启策略
         */
//        env.setRestartStrategy(RestartStrategies.noRestart());
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                //失败重启次数
                3,
                //重启间隔时间
                Time.seconds(5)
        ));

//        env.setRestartStrategy(RestartStrategies.failureRateRestart(3,Time.minutes(5),Time.seconds(3)));


        //TODO 2.读取数据源
        DataStream<Order> orderDataStream = env.addSource(new CustomerOrder.CustomerOrderSource());

        //设置wm
        orderDataStream.assignTimestampsAndWatermarks(WatermarkStrategy
                .<Order>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner((order, timestamp) -> order.getTimeStamp()));

        //TODO 3.转换操作

        //TODO 4.sink操作
        orderDataStream.print();

        //TODO 5.执行程序
        env.execute();
    }
}
